package com.zihan.evm;

import com.alibaba.otter.canal.protocol.FlatMessage;
import com.zihan.evm.annotation.CanalTable;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
import javax.annotation.Resource;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @author yang peng
 * @date 2019/3/2713:33
 */
@Component
@Slf4j
public class HandlerUtil {
    @Resource
    HandlerFactory handlerFactory;

    public static Map<String, EntryHandler> getTableHandlerMap(List<? extends EntryHandler> entryHandlers) {
        Map<String, EntryHandler> map = new ConcurrentHashMap<>();
        if (entryHandlers != null && entryHandlers.size() > 0) {
            for (EntryHandler handler : entryHandlers) {
                String canalTableName = getCanalTableName(handler);
                if (canalTableName != null) {
                    map.putIfAbsent(canalTableName.toLowerCase(), handler);
                } else {
                    String name = GenericUtil.getTableGenericProperties(handler);
                    if (name != null) {
                        map.putIfAbsent(name.toLowerCase(), handler);
                    }
                }
            }
        }
        return map;
    }



    public static String getCanalTableName(EntryHandler entryHandler) {
        CanalTable canalTable = entryHandler.getClass().getAnnotation(CanalTable.class);
        if (canalTable != null) {
            return canalTable.value();
        }
        return null;
    }


    /**
     * 消息消费
     * @param info
     * @param <T>
     */
    public <T> void handleMessage(FlatMessage info){
        log.info("canal消息",info);
        String table = info.getTable();
        String type = info.getType();
        EntryHandler<T> handler = handlerFactory.getHandler(table);
        if (null == handler){
            log.error("该表不在同步范围：{}", table);
            return;
        }

        List<Map<String, String>> data = info.getData();
        if (data == null || data.size() == 0) {
            log.warn("解析数据失败");
            return;
        }

        for (int i = 0; i < data.size(); i++) {
            T model = handlerFactory.getModel(info.getData(), handler, i);
            if ("INSERT".equals(type)){
                handler.insert(model);
            }else if ("UPDATE".equals(type)){
                T old = handlerFactory.getModel(info.getOld(), handler, i);
                handler.update(old, model);
            }else if ("DELETE".equals(type)){
                handler.delete(model);
            }
        }
    }
}
